package com.yuebook.smartrpc.framework.invoker;

import com.yuebook.smartrpc.framework.cluster.ClusterStrategy;
import com.yuebook.smartrpc.framework.helper.PropertyConfigHelper;
import com.yuebook.smartrpc.framework.model.ProviderService;
import com.yuebook.smartrpc.framework.model.SmartRpcRequest;
import com.yuebook.smartrpc.framework.model.SmartRpcResponse;
import com.yuebook.smartrpc.framework.serialization.NettyDecodeHandler;
import com.yuebook.smartrpc.framework.serialization.NettyEncodeHandler;
import com.yuebook.smartrpc.framework.serialization.common.SerializerType;
import com.yuebook.smartrpc.framework.zookeeper.IRegisterCenterInvoker;
import com.yuebook.smartrpc.framework.zookeeper.RegisterCenter;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

/**
 * 消费者bean代理工厂
 * @author honglei
 * @date 2018/11/18 下午9:13
 */
public class InvokerProxyBeanFactory implements InvocationHandler {

    private Class<?> targetInterface;

    private long timeout;

    private SerializerType serializerType = PropertyConfigHelper.getSerializeType();

    private InvokerProxyBeanFactory(Class<?> targetInterface, long timeout) {
        this.targetInterface = targetInterface;
        this.timeout = timeout;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        final SmartRpcRequest request = new SmartRpcRequest();
        ProviderService providerService = new ProviderService();
        providerService.setServiceItf(targetInterface);
        providerService.setServiceMethod(method);
        request.setProviderService(providerService);
        request.setInvokeTimeout(timeout);
        request.setInvokedMethodName(method.getName());
        request.setArgs(args);
        request.setUniqueKey(UUID.randomUUID().toString());

        //选择合适的服务
        IRegisterCenterInvoker registerCenterInvoker = RegisterCenter.singleton();
        ClusterStrategy clusterStrategy = PropertyConfigHelper.getClusterStrategy();
        final ProviderService remoteProviderService = clusterStrategy.select(registerCenterInvoker.getServiceMetaData().get(targetInterface.getName()));

        FutureTask<Object> task = new FutureTask<>(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                InetSocketAddress address = new InetSocketAddress(remoteProviderService.getServiceIp(), remoteProviderService.getServicePort());
                NioEventLoopGroup group = new NioEventLoopGroup();
                Bootstrap bootstrap = new Bootstrap();
                try {
                    bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .remoteAddress(address)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                socketChannel.pipeline().addLast(new NettyDecodeHandler(SmartRpcResponse.class, serializerType));
                                socketChannel.pipeline().addLast(new NettyEncodeHandler(serializerType));
                                socketChannel.pipeline().addLast(new NettyClientInvokeInHandler(request));
                            }
                        });
                    ChannelFuture future = bootstrap.connect().sync();
                    future.channel().closeFuture().sync();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    group.shutdownGracefully().sync();
                }
                SmartRpcResponse response = (SmartRpcResponse) InvokerResponseHolder.getResponse(request.getUniqueKey());
                InvokerResponseHolder.removeResponse(request.getUniqueKey());
                return response.getResult();
            }
        });
        new Thread(task).start();
        return task.get(10000000, TimeUnit.MILLISECONDS);
    }

    public Object getProxy() {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{targetInterface}, this);
    }

    private static InvokerProxyBeanFactory singleton;

    public static InvokerProxyBeanFactory singleton(Class<?> targetInterface, long timeout){
        if (singleton == null) {
            synchronized (InvokerProxyBeanFactory.class) {
                if (singleton == null) {
                    singleton = new InvokerProxyBeanFactory(targetInterface, timeout);
                }
            }
        }
        return singleton;
    }
}
